﻿//#define TRACE_MESSAGE_CHANGE_STATE

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using System.Threading;
using System.Diagnostics;

/*
 * Message is in the core part of the Messaging framework
 * 
 * Message represents a unit of a work to be performed
 * 
 */

namespace Bohdan.Arch.Messaging
{
    /// <summary>
    /// MessageState describes where the message is in the processing queue
    /// 
    /// You can only place a message which is in a ReadyOrProcessed state (its default state)
    /// 
    /// It is useful to help reveal problems connected with posting the same message several times
    /// </summary>
    public enum MessageState
    {
        ReadyOrProcessed,
        PlacedInQueue,
        BeingProcessed,
        Cancelled,
    }

    public abstract class GeneralMessage : IComparable<GeneralMessage>
    {
        /// <summary>
        /// Message Id.
        /// 
        /// very useful for tracking the message being executed
        /// </summary>
        protected Guid _Id = Guid.NewGuid();
        protected volatile MessageState _State = MessageState.ReadyOrProcessed;

        public virtual Guid MessageId
        {
            get
            {
                return _Id;
            }
        }
        public MessagingBus Bus { get; set; }
        public abstract Guid TargetSubscriberId { get; }
        public object Tag { get; set; }
        
        protected EventWaitHandle _ewhProcessed = new EventWaitHandle(false, EventResetMode.ManualReset);
        protected Queue<GeneralMessage> _waitingQueue = new Queue<GeneralMessage>();
        protected int _dependentMessages = 1;

        protected object _syncObj = new object();

        protected DateTime _PlacedInQueue;
        protected DateTime _StartedProcessing;
        protected DateTime _FinishedProcessing;

        public virtual float Priority 
        { 
            get {
                return 1.0f;
            }
        }

        public Action<GeneralMessage> BeforeExecution;
        public Action<GeneralMessage> AfterExecution;

        public GeneralMessage()
        {
        }

        public DateTime PlacedInQueue
        {
            get
            {
                return _PlacedInQueue;
            }
            internal set
            {
                _PlacedInQueue = value;
            }
        }
        public DateTime StartedProcessing
        {
            get
            {
                return _StartedProcessing;
            }
            internal set
            {
                _StartedProcessing = value;
            }
        }
        public DateTime FinishedProcessing
        {
            get
            {
                return _FinishedProcessing;
            }
            internal set
            {
                _FinishedProcessing = value;
            }
        }

        /// <summary>
        /// This method is called by the Bus when message is placed in queue
        /// </summary>
        internal void ReportPlacedInQueue()
        {
            lock (_syncObj)
            {
                if (_State != MessageState.ReadyOrProcessed)
                {
#if TRACE_MESSAGE_CHANGE_STATE
                    Trace.TraceError("Attempting to place in queue a message being processed.", MessageId);
#endif
                    throw new ArgumentException("Attempting to place in queue a message being processed.");
                }
                _State = MessageState.PlacedInQueue;
#if TRACE_MESSAGE_CHANGE_STATE
                Trace.TraceInformation("Message has been placed in queue. id = {0}", MessageId);
#endif
            }
        }

        /// <summary>
        /// This method is called by subscriber when the message is about to be processed
        /// </summary>
        internal void ReportBeingProcessed()
        {
            lock (_syncObj)
            {
                if (_State != MessageState.PlacedInQueue)
                {
#if TRACE_MESSAGE_CHANGE_STATE
                    Trace.TraceError("Attempting to process a message not being placed in queue. id = {0}", MessageId);
#endif
                    throw new ArgumentException("Attempting to process a message not being placed in queue.");
                }
                _State = MessageState.BeingProcessed;
#if TRACE_MESSAGE_CHANGE_STATE
                Trace.TraceInformation("Message is being processed. id = {0}", MessageId);
#endif
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        public virtual bool PreExecutionValidate()
        {
            return true;
        }

        public virtual bool IsProcessed
        {
            get
            {
                lock (_syncObj)
                    return _State == MessageState.ReadyOrProcessed || _State == MessageState.Cancelled;
            }
        }

        public virtual void AddToWaitingQueue(params GeneralMessage[] msgs)
        {
            lock (_waitingQueue)
                foreach (var msg in msgs)
                {
                    Interlocked.Increment(ref _dependentMessages);
                    msg.OnMessageProcessed += DecDependent;
                    _waitingQueue.Enqueue(msg);
                }
        }

        /// <summary>
        /// Waits for completion of the current message and all the dependent messages
        /// </summary>
        /// <returns>true</returns>
        public virtual bool WaitCompleted()
        {
            return WaitCompleted(TimeSpan.FromMilliseconds(int.MaxValue - 2));
        }

        /// <summary>
        /// Waits for completion of the current message and all the dependent messages for a limited time span
        /// </summary>
        /// <param name="ts">time span allowed for waiting</param>
        /// <returns>true if the the message completed</returns>
        public virtual bool WaitCompleted(TimeSpan ts)
        {
            if (!WaitAndCorrect(_ewhProcessed, ref ts))
                return false;

            int prevCount = 0;
            while (true)
            {
                GeneralMessage[] messagesToWait = null;

                lock (_waitingQueue)
                    messagesToWait = _waitingQueue.ToArray();

                if (prevCount == messagesToWait.Length)
                    break;

                prevCount = messagesToWait.Length;

                foreach (GeneralMessage msg in messagesToWait)
                    if (!WaitAndCorrect(msg, ref ts))
                        return false;
            }

            return true;
        }

        #region Helper method to wait for smth and decrease the amount of time available for waiting
        private static bool WaitAndCorrect(EventWaitHandle ewh, ref TimeSpan ts)
        {
            DateTime dtBefore = DateTime.Now;
            bool res = ewh.WaitOne(ts);
            TimeSpan tsElapsed = DateTime.Now - dtBefore;
            ts = (ts >= tsElapsed) ? ts - tsElapsed : TimeSpan.Zero;
            return res;
        }

        private static bool WaitAndCorrect(GeneralMessage msg, ref TimeSpan ts)
        {
            DateTime dtBefore = DateTime.Now;
            bool res = msg.WaitCompleted(ts);
            TimeSpan tsElapsed = DateTime.Now - dtBefore;
            ts = (ts >= tsElapsed) ? ts - tsElapsed : TimeSpan.Zero;
            return res;
        }
        #endregion

        internal void BeforeExecutionFire()
        {
            if (BeforeExecution != null)
                BeforeExecution(this);
        }
        internal void AfterExecutionFire()
        {
            if (AfterExecution != null)
                AfterExecution(this);
        }

        protected event EventHandler _OnMessageProcessed;
        protected object _OnMessageProcessedSyncObj = new object();
        public event EventHandler OnMessageProcessed
        {
            add
            {
                lock (_OnMessageProcessedSyncObj)
                    _OnMessageProcessed += value;
            }
            remove
            {
                lock (_OnMessageProcessedSyncObj)
                    _OnMessageProcessed -= value;
            }
        }

        protected void DecDependent(object sender, EventArgs e)
        {
            if (Interlocked.Decrement(ref _dependentMessages) == 0)
            {
                lock (_OnMessageProcessedSyncObj)
                {
                    if (_OnMessageProcessed == null)
                        return;

                    _OnMessageProcessed(this, EventArgs.Empty);
                }
            }
        }

        public virtual void DesclareMessageProcessed()
        {
            lock (_syncObj)
            {
                if (_State != MessageState.BeingProcessed)                    // That's ok - PreValidate was false
                    _State = MessageState.Cancelled;
                    //throw new ArgumentException("Improper state of the message in DesclareMessageProcessed()");
                else
                    _State = MessageState.ReadyOrProcessed;
#if TRACE_MESSAGE_CHANGE_STATE
                Trace.TraceInformation("Message is reported as processed. id = {0}", MessageId);
#endif
            }

            _ewhProcessed.Set();

            DecDependent(this, EventArgs.Empty);
        }

        public override string ToString()
        {
            return GetType().Name + " (" + MessageId.ToString() + ")";
        }

        #region IComparable<GeneralMessage> Members

        public int CompareTo(GeneralMessage other)
        {
            int t = Priority.CompareTo(other.Priority);
            if (t != 0)
                return t;
            return MessageId.CompareTo(other.MessageId);
        }

        #endregion
    }
}
